#pragma once
#include <iostream>
#include <log.h>
#include <memory>
#include <librdkafka/rdkafkacpp.h>
// #include <librdkafka/rdkafka.h>
#include <mutex>
#include <string>

/*
topic（主题）: 类似于数据库中的表，通常将相同类型消息存储再一个topic中，mysql中的表是结构化的，
topic是半结构化的，所以不同类型的消息也可以存放在一个主题中
partition（分区）：主题可以包含多个分区，kafka将主题拆分成不同的分区，并且存放再不同的服务器上
这样就具有了扩展性
record（消息记录）：kafka中的消息，是以键值对（kv）的方式进行存储的，如果发送消息的时候，如果key
为空，那就会以轮询的方式将消息写到不同的分区中，key不为空，相同key的消息会存储到同一个分区
replication-factor（副本因子）：分区如果只存在一份，分区损坏，那么信息就会丢失，所以kafka引入副本
因子，为3，表示主分区有1个（leader），对这个主分区的备份有2个（follower），所有分区的写入操作都是写入到主中，读取
也是从主中读，follower只负责从主分区中同步数据，kafaka会查看不同分区的同步状态
broker（消息代理）：kafka是由多个broker组成的，broker负责数据的读写请求，并将数据写入到磁盘中，
通常每个服务器上启动一个broker的实例



*/

namespace lyt{
    class KafkaProducer{
    private:
        // Kafka 生产者客户端对象，用于实际发送消息,生产者对象是由 KafkaProducer 独占管理的，
        // 生命周期明确，用不到共享语义,所以不适用shared_ptr
        std::unique_ptr<RdKafka::Producer> _producer;
        // Kafka Topic 对象，表示目标 topic（已绑定生产者）
        std::unique_ptr<RdKafka::Topic> _topic;
        std::string _topic_name;
        // 为了线程安全，对发送操作加锁，避免多线程竞争
        std::mutex _mutex;
    public:
        using ptr = std::shared_ptr<KafkaProducer>;

        // - brokers(消息代理): Kafka 集群的地址列表，格式如 "localhost:9092,broker2:9092"
        // - topic: 要发送消息的 Kafka topic 名称
        KafkaProducer(const std::string& brokers,const std::string& topic_name)
                    :_topic_name(topic_name){
            // 存储错误信息
            std::string errstr;
            // 创建全局配置对象（CONF_GLOBAL），用于配置生产者的全局属性
            RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
            // 设置 Kafka broker 地址（即 Kafka 集群的入口），"localhost:9092",必须设置，不然客户端无法知道将消息发送到哪里
            conf->set("bootstrap.servers",brokers,errstr);
            // 创建生产者实例，绑定配置，失败时返回nullptr，并填充errstr
            _producer.reset(RdKafka::Producer::create(conf, errstr));
            if(!_producer){
                std::cerr << "Kafka producer create failed: " << errstr << std::endl;
                throw std::runtime_error("Kafka init failed");
            }
            // conf 是使用 new 创建的，Kafka Producer 接管后可以释放掉
            delete conf;
            // 创建topic配置对象
            RdKafka::Conf* topic_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
            // 创建topic对象，绑定到producer上的某个topic名称
            _topic.reset(RdKafka::Topic::create(_producer.get(), _topic_name, topic_conf, errstr));
            if (!_topic) {
                std::cerr << "Kafka topic create failed: " << errstr << std::endl;
                throw std::runtime_error("Kafka topic init failed");
                // 同样释放 topic_conf，创建 Topic 后不再需要
                delete topic_conf;
            }
        }
        // 确保kafka producer资源释放前将缓存中的消息发送出去
        ~KafkaProducer(){
            if(_producer){
                // flush 会等待3s将所有再队列中的消息发送给broker
                _producer->flush(3000);
            }
        }

        // - payload: 实际要发送的数据（消息体）
        // - true 表示消息已成功投递至 Kafka 客户端（但不一定已写入 broker）
        bool send(const std::string& key,const std::string& payload){
            // 加锁，防止多线程同事调用produce导致竞争（librdkafka 本身是线程安全的，但应用层可以保证逻辑一致性）
            
            std::unique_lock<std::mutex> lock(_mutex);
            // 发送kafka消息
            RdKafka::ErrorCode err = _producer->produce(

                _topic.get(),                             // 目标主题
                RdKafka::Topic::PARTITION_UA,       // 使用自动分区
                RdKafka::Producer::RK_MSG_COPY,     // 拷贝消息内容，而不是使用指针（安全，避免 dangling 指针）
                const_cast<char*>(payload.c_str()),payload.size(),   // 消息体内容
                &key,                               // 消息key，用于分区逻辑
                nullptr);                           // 传递自定义的opaque指针，通常用于向上回调
            // 如果 produce 失败，则打印错误日志
            if (err != RdKafka::ERR_NO_ERROR) {
                std::cerr << "Kafka produce failed: " << RdKafka::err2str(err) << std::endl;
                return false;
            }
            // 触发 Kafka 客户端内部事件处理，例如 delivery report 回调、错误回调等
            // poll(0) 表示非阻塞，只处理已有事件
            _producer->poll(0);
            return true;
        }
    
    };

    
}
